package org.kafka.metrics;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.function.Consumer;

import org.apache.kafka.clients.CommonClientConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.admin.AdminClient;
import kafka.admin.AdminClient.ConsumerSummary;
import kafka.admin.ConsumerGroupCommand.ConsumerGroupCommandOptions;
import kafka.admin.ConsumerGroupCommand.KafkaConsumerGroupService;
import kafka.coordinator.GroupOverview;
import scala.Function1;
import scala.Option;
import scala.Tuple2;
import scala.collection.Seq;
import scala.collection.JavaConversions;

public class KafkaLagService_1001 {
    private static Logger log = LoggerFactory.getLogger(KafkaLagService_1001.class);
    AdminClient adminClient;
    public void init() {
        Properties props = new Properties();
        props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093");
        adminClient = AdminClient.create(props);
    }

//    public List<String> listConsumergroup() {
//        List<String> options = new ArrayList<>();
//        Collection<GroupOverview> groups = JavaConversions.asJavaCollection(adminClient.listAllConsumerGroupsFlattened());
//        for (GroupOverview groupOverview : groups) {
//            options.add(groupOverview.groupId());
//        }
//        return options;
//    }
//
//    public GroupInfo describe(String group) {
//        scala.collection.immutable.List<ConsumerSummary> consumerSummaries = adminClient.describeConsumerGroup(group);
//
//
//
//
//        ConsumerGroupCommandOptions cgOptions = new ConsumerGroupCommandOptions(options.toArray(new String[0]));
//        System.out.println(cgOptions);
//        Tuple2<Option<String>, Option<Seq<PartitionAssignmentState>>> info;
//        try {
//            consumerGroupService = new KafkaConsumerGroupService(cgOptions);
//            info = consumerGroupService.describeGroup();
//            if (info._2().nonEmpty()) {
//                GroupInfo groupInfo = new GroupInfo().setState(info._1().get());
//                List<PartitionAssignmentStateBO> result = new ArrayList<>();
//                List<PartitionAssignmentState> list = JavaConversions.seqAsJavaList(info._2().get());
//                for (PartitionAssignmentState partitionAssignmentState : list) {
//                    PartitionAssignmentStateBO partInfo = new PartitionAssignmentStateBO();
//                    copy(partInfo, partitionAssignmentState);
//                    result.add(partInfo);
//                    log.info("pt{},lag{},offset{},logend{}", partInfo.partition, partInfo.lag, partInfo.offset,
//                            partInfo.logEndOffset);
//                }
//                groupInfo.setPatitions(result);
//                return groupInfo;
//            } else {
//                return new GroupInfo().setState(info._1().get());
//            }
//        } finally {
//            if (null != consumerGroupService) {
//                consumerGroupService.close();
//            }
//        }
//    }
//
//    public static void copy(PartitionAssignmentStateBO to, PartitionAssignmentState from) {
//        if (null == to || from == null) {
//            throw new IllegalStateException("from or to is null");
//        }
//        to.setClientID(from.clientId().get());
//        to.setGroup(from.group());
//        to.setTopic(from.topic().get());
//        to.setConsumerID(from.consumerId().get());
//        to.setLag((Long) from.lag().get());
//        to.setLogEndOffset((Long) from.logEndOffset().get());
//        to.setOffset((Long) from.offset().get());
//        to.setPartition((Integer) from.partition().get());
//        if (from.coordinator().nonEmpty()) {
//            to.setCoordinator("" + from.coordinator().get().id());
//        }
//    }
}
